/*
 * Copyright 2015 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.benmei.ntlive.socket.server.netty.handler;

import com.benmei.ntlive.common.CourseType;
import com.benmei.ntlive.common.UserRole;
import com.benmei.ntlive.constant.Constants;
import com.benmei.ntlive.dto.*;
import com.benmei.ntlive.dto.ntmsg.cons.Action;
import com.benmei.ntlive.entity.*;
import com.benmei.ntlive.exception.NtClientException;
import com.benmei.ntlive.service.LiveService;
import com.benmei.ntlive.utils.JsonUtil;
import com.benmei.ntlive.vo.Room;
import com.benmei.ntlive.vo.Session;
import com.benmei.ntlive.vo.SessionRepository;
import com.benmei.ntlive.vo.User;
import io.netty.channel.*;
import io.netty.util.concurrent.ScheduledFuture;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * NtliveServerHandler
 *
 * @author Peter
 * @date 2017-08-17
 */
@Component
@Qualifier("ntliveServerHandler")
@ChannelHandler.Sharable
public class NtliveServerHandler extends ChannelHandlerAdapter {
    private static Logger logger = LoggerFactory.getLogger(NtliveServerHandler.class);

    @Autowired
    private LiveService liveService;

    @Autowired
    private SessionRepository sessionRepository;

    private static Map<String, ScheduledFuture> taskMap = new ConcurrentHashMap<String, ScheduledFuture>();

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelActive();
        String client = ctx.channel().remoteAddress().toString();
        logger.debug("A client connect success for {}", client);

        ctx.writeAndFlush(JsonUtil.toJson(new TcpResponse("connect")) + Constants.delimiter);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String stringMessage = (String) msg;
        try {
            // JSON格式校验
            if (!JsonUtil.isValidJSON(stringMessage)) {
                throw new NtClientException("Invalid JSON data format");
            }
            // 头部校验
            TcpRequest req = JsonUtil.toObject(stringMessage, TcpRequest.class);
            validateTcpHead(req.getHead());

            // action 校验
            Map<String, Object> body = req.getBody();
            Object objAction = body.get("action");
            if (objAction == null) {
                throw new NtClientException("Invalid data format:action is required");
            }
            if (!String.class.isInstance(objAction)) {
                throw new NtClientException("Invalid data format:action must be string");
            }
            String action = (String) objAction;
            if (!Action.isValidateAction(action)) {
                throw new NtClientException("Invalid action value '" + action + "'");
            }

            // 收到心跳消息，Pong
            if (Action.HEART_BEAT.getName().equalsIgnoreCase(action)) {
//                TcpResponse heatBeat = new TcpResponse(Action.HEART_BEAT.getName());
//                heatBeat.setMessage("Pong");
//                String heatBeatMsg = JsonUtil.toJson(heatBeat);
//                logger.trace("Response heart beat messsage to client ---> " + heatBeatMsg);
//                ctx.channel().writeAndFlush(heatBeatMsg + Constants.delimiter);
            } else {
                // 检查是否有session，没有就创建。
                createSession(ctx, req, action);
                // 业务逻辑处理
                liveService.process(ctx, req, action);
            }
        } catch (NtClientException e) {
            logger.error(e.getMessage(), e);
            ctx.channel().writeAndFlush(e.toJson() + Constants.delimiter);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            TcpResponse response = new TcpResponse(Action.CREATE_SESSION.getName());
            response.setMessage("Server Error");
            response.setStatus(500);

            ctx.channel().writeAndFlush(JsonUtil.toJson(response) + Constants.delimiter);
        }
    }

    private Integer getIntegerForRoomId(String action, Object objRoomId) throws NtClientException {
        Integer roomId = null;
        try {
            roomId = Integer.parseInt(objRoomId.toString());
        } catch (NumberFormatException e) {
            logger.error("RoomId must be number, roomId=" + objRoomId.toString(), e);
            throw new NtClientException("RoomId must be number, roomId=" + objRoomId.toString(), action);
        }
        return roomId;
    }

    private void createSession(ChannelHandlerContext ctx, TcpRequest req, String action) throws NtClientException {
        Session oldSession = sessionRepository.getSessionByKey(ctx.channel().remoteAddress().toString());
        // 如果有session了，直接返回，否则创建session
        if (oldSession != null) {
            return;
        }

        // 校验roomId是否有效
        final Integer roomId = req.getHead().getRoomId();
        if (req.getHead().getCourseType().equals(CourseType.c_1v1.getName())) {
            final ClassRoom classRoom = liveService.findRoomById(roomId);
            if (classRoom == null) {
                throw new NtClientException("ClassRoom do not exists, The roomId=" + roomId, action);
            }
        } else {
            final McCourseSchedule mcCourseSchedule = liveService.findMcCourseRoomById(roomId);
            if (mcCourseSchedule == null) {
                throw new NtClientException("McCourseRoom do not exists, The roomId=" + roomId, action);
            }
        }
        String courseTypeName = req.getHead().getCourseType();
        int _courseType = CourseType.getByName(courseTypeName).getCode();
        int role = req.getHead().getNtClientRole();
        String token = req.getHead().getNtToken();
        Integer clientOs = req.getHead().getNtClientOSType();
        Channel channel = ctx.channel();

        // 创建教室
        Room roomSession = sessionRepository.getRoom(courseTypeName, roomId);
        if (roomSession == null) {
            // 创建教室级别的session
            roomSession = sessionRepository.createRoom(courseTypeName, roomId);
            logger.info("Created roomSession，endpoint:{}", channel.remoteAddress().toString());

            // 记录创建教室的日志
            ClassRoomEntryLog roomLog = new ClassRoomEntryLog();
            roomLog.setRoom_id(roomId);
            roomLog.setEvent_type("createTcpRoomSession");
            roomLog.setEvent_date(new Date());
            roomLog.setUser_role(role);
            roomLog.setClient_os_type(clientOs);
            roomLog.setDetail("Created TCP Room session with " + role + "endPoint:" + channel.remoteAddress().toString() + " for os:" + clientOs);
            liveService.roomLog(roomLog);
        }

        if (role == Constants.UserRole.STUDENT) {
            Member member = liveService.getMemberByToken(token);
            if (member == null) {
                throw new NtClientException("Invalid token:" + token + " for member", action);
            }
            Integer agora_uid = liveService.getAgoraUid(AgoraUid.Role.student, member.getMemb_id());
            User user = new User(member, clientOs, agora_uid);
            // 记录session创建日志
            ClassRoomEntryLog roomLog = new ClassRoomEntryLog();
            roomLog.setRoom_id(roomId);
            roomLog.setEvent_type("createTcpSession");
            roomLog.setEvent_date(new Date());
            roomLog.setUser_id(user.getId());
            roomLog.setUser_role(role);
            roomLog.setClient_os_type(clientOs);
            roomLog.setCourse_type(_courseType);
            final Session oldStudent = roomSession.getStudentByMembId(member.getMemb_id());
            if (oldStudent == null) {
                logger.info("Created TCP room session for student:(id:{}-name:{}-token:{})，roomId={}, endpoint:{}", user.getId(), user.getName(), user.getToken(), roomId, channel.remoteAddress().toString());
                roomLog.setDetail("Created TCP room session(Student),endPoint:" + channel.remoteAddress().toString() + " for OS:" + clientOs);
            } else {
                logger.info("Created other TCP room session for student:(id:{}-name:{}-token:{})，roomId={}, endpoint:{}", user.getId(), user.getName(), user.getToken(), roomId, channel.remoteAddress().toString());
                roomLog.setDetail("Created other TCP room session(Student),endPoint:" + channel.remoteAddress().toString() + " for OS:" + clientOs);
                logger.info("old Student Device endpoint [{}]", oldStudent.getChannel().remoteAddress().toString());
                logger.info("new Student Device endpoint [{}]", channel.remoteAddress().toString());
                // 发消息，通知上一个客户端退出系统
                TcpResponse response = new TcpResponse(Action.LOGOUT.getName());
                response.setMessage("您已在新的设备登录，该设备退出登录。");
                oldStudent.getChannel().writeAndFlush(JsonUtil.toJson(response) + Constants.delimiter).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (future.isSuccess()) {
                            logger.info("通知退出成功 to: " + oldStudent.getChannel().remoteAddress() + " User:" + JsonUtil.toJson(oldStudent.getUser()));
                        } else {
                            logger.error("通知退出失败 to: " + oldStudent.getChannel().remoteAddress() + " 原因为:" + future.cause() + " User:" + JsonUtil.toJson(oldStudent.getUser()));
                        }
                    }
                });
                oldStudent.getChannel().close();// 关闭旧的连接
            }

            Session session = new Session(channel, user);
            roomSession.addStudent(session);
            liveService.roomLog(roomLog);

            // session 创建完成后给客户端回消息：举手列表
            final String clientEndpoint = ctx.channel().remoteAddress().toString();
            ctx.channel().writeAndFlush(JsonUtil.toJson(new TcpResponse(action)) + Constants.delimiter);
            // 向新连接的用户发送教室举手列表
            ctx.channel().writeAndFlush(JsonUtil.toJson(new HandsUpListResponse(roomSession)) + Constants.delimiter).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        logger.info("向新连接的用户发送教室举手列表 成功 to: " + clientEndpoint);
                    } else {
                        logger.error("向新连接的用户发送教室举手列表 失败 to: " + clientEndpoint + " 原因为:" + future.cause());
                    }
                }
            });

            // session 创建完成后给客户端回消息：发送当前教室学生数
            sessionRepository.sendStudentSizeMessageForRoom(roomSession);// 发送当前教室学生数

            // session 创建完成后给客户端回消息：星星数量
            responseStarMsg(ctx, req, action, roomId);

            // 当握手成功后，Login响应向下透传，主动发送心跳消息
            //NioEventLoop是一个Schedule,因此支持定时器的执行，创建心跳计时器
            //使用定时任务发送心跳消息
            ScheduledFuture<?> heartBeat = ctx.executor().scheduleAtFixedRate(new NtliveServerHandler.HeartBeatTask(ctx), 0, 60000, TimeUnit.MILLISECONDS);
            String taskKey = ctx.channel().remoteAddress().toString();
            taskMap.put(taskKey, heartBeat);
        } else if (role == Constants.UserRole.TEACHER) {
            Teacher teacher = liveService.getTeacherByToken(token);
            if (teacher == null) {
                throw new NtClientException("Invalid token:" + token + " for teacher", action);
            }
            Integer agora_uid = liveService.getAgoraUid(AgoraUid.Role.teacher, teacher.getTea_id());
            User user = new User(teacher, clientOs, agora_uid);
            // 记录session创建日志
            ClassRoomEntryLog roomLog = new ClassRoomEntryLog();
            roomLog.setRoom_id(roomId);
            roomLog.setEvent_type("createTcpSession");
            roomLog.setEvent_date(new Date());
            roomLog.setUser_id(user.getId());
            roomLog.setUser_role(role);
            roomLog.setClient_os_type(clientOs);
            roomLog.setCourse_type(_courseType);
            final Session oldTeacher = roomSession.getTeacherByTeaId(teacher.getTea_id());
            if (oldTeacher == null) {
                logger.info("Created TCP Teacher session for teacher:(id:{}-name:{}-token:{})，roomId={}, endpoint:{}", user.getId(), user.getName(), user.getToken(), roomId, channel.remoteAddress().toString());
                roomLog.setDetail("Created TCP Teacher session(Teacher),endPoint:" + channel.remoteAddress().toString() + " for os:" + clientOs);
            } else {
                logger.info("Created other TCP Teacher session for teacher:(id:{},name:{},token:{})，roomId={}, endpoint:{}", user.getId(), user.getName(), user.getToken(), roomId, channel.remoteAddress().toString());
                logger.info("old teacher endpoint [{}]", oldTeacher.getChannel().remoteAddress().toString());
                logger.info("new teacher endpoint [{}]", channel.remoteAddress().toString());
                roomLog.setDetail("Created other TCP Teacher session(Teacher),endPoint:" + channel.remoteAddress().toString() + " for os:" + clientOs);
                // 发消息，通知上一个客户端退出系统
                TcpResponse response = new TcpResponse(Action.LOGOUT.getName());
                response.setMessage("您已在新的设备登录，该设备退出登录。");
                oldTeacher.getChannel().writeAndFlush(JsonUtil.toJson(response) + Constants.delimiter).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (future.isSuccess()) {
                            logger.info("通知退出成功 to: " + oldTeacher.getChannel().remoteAddress() + " User:" + JsonUtil.toJson(oldTeacher.getUser()));
                        } else {
                            logger.error("通知退出失败 to: " + oldTeacher.getChannel().remoteAddress() + " 原因为:" + future.cause() + " User:" + JsonUtil.toJson(oldTeacher.getUser()));
                        }
                    }
                });

            }

            Session session = new Session(channel, user);
            roomSession.addTeacher(session);
            liveService.roomLog(roomLog);

            // session 创建完成后给客户端回消息
            final String clientEndpoint = ctx.channel().remoteAddress().toString();
            ctx.channel().writeAndFlush(JsonUtil.toJson(new TcpResponse(action)) + Constants.delimiter);
            // 向新连接的用户发送教室举手列表
            ctx.channel().writeAndFlush(JsonUtil.toJson(new HandsUpListResponse(roomSession)) + Constants.delimiter).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        logger.info("向新连接的用户发送教室举手列表 成功 to: " + clientEndpoint);
                    } else {
                        logger.error("向新连接的用户发送教室举手列表 失败 to: " + clientEndpoint + " 原因为:" + future.cause());
                    }
                }
            });

            // 发送当前教室学生数
            sessionRepository.sendStudentSizeMessageForRoom(roomSession);

            // session 创建完成后给客户端回消息：星星数量
            responseStarMsg(ctx, req, action, roomId);

            // 当握手成功后，Login响应向下透传，主动发送心跳消息
            //NioEventLoop是一个Schedule,因此支持定时器的执行，创建心跳计时器
            //使用定时任务发送心跳消息
            ScheduledFuture<?> heartBeat = ctx.executor().scheduleAtFixedRate(new NtliveServerHandler.HeartBeatTask(ctx), 0, 60000, TimeUnit.MILLISECONDS);
            String taskKey = ctx.channel().remoteAddress().toString();
            taskMap.put(taskKey, heartBeat);
//                ctx.executor().shutdownGracefully();
            logger.info(" \n       start schedule task ------》》》》》》》》" + heartBeat.toString());
        } else {
            throw new NtClientException("Invalid ntClientRole", action);
        }
    }

    // 刚进入教室时回复星星数量
    private void responseStarMsg(ChannelHandlerContext ctx, TcpRequest req, String action, final Integer roomId) {
        final String courseType = req.getHead().getCourseType();
        if (courseType.equals(CourseType.c_1v1.getName())) {
            Integer star_size = liveService.findStarByRoomId(roomId);
            if (star_size == null) {
                star_size = 0;
            }
            final Integer starSize = star_size;
            final StarResponse starResponse = new StarResponse(action, starSize);
            ctx.channel().writeAndFlush(JsonUtil.toJson(starResponse) + Constants.delimiter).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        logger.info("发送星星给老师成功（教室：" + courseType + "_" + roomId + "）starSize=" + starSize);
                    } else {
                        logger.error("发送星星给老师失败（教室：" + courseType + "_" + roomId + "）starSize=" + starSize);
                    }
                }
            });
        }
    }

    private void validateTcpHead(TcpRequestHead head) throws NtClientException {
        if (head == null) {
            throw new NtClientException("Invalid data format:head is required");
        }
        String apiVersion = head.getNtApiVersion();
        if (StringUtils.isBlank(apiVersion)) {
            throw new NtClientException("Invalid data format:ntApiVersion is required");
        }

        Long timestamp = head.getNtTimestamp();
        if (timestamp == null) {
            throw new NtClientException("Invalid data format:ntTimestamp is required");
        }

        String language = head.getNtClientLanguage();
        if (StringUtils.isBlank(language)) {
            throw new NtClientException("Invalid data format:ntClientLanguage is required");
        }

        String signature = head.getNtSignature();
        if (StringUtils.isBlank(signature)) {
            throw new NtClientException("Invalid data format:ntSignature is required");
        }

        String token = head.getNtToken();
        if (StringUtils.isBlank(token)) {
            throw new NtClientException("Invalid data format:ntToken is required");
        }

        Integer osType = head.getNtClientOSType();
        if (osType == null) {
            throw new NtClientException("Invalid data format:ntClientOSType is required");
        }

        Integer clientType = head.getNtClientType();
        if (clientType == null) {
            throw new NtClientException("Invalid data format:ntClientType is required");
        }

        Integer role = head.getNtClientRole();
        if (role == null) {
            throw new NtClientException("Invalid data format:ntClientRole is required");
        }
        if (!UserRole.contains(role)) {
            throw new NtClientException("ntClientRole must be 1 or 2, ntClientRole:" + role);
        }

        String courseType = head.getCourseType();
        if (StringUtils.isBlank(courseType)) {
            throw new NtClientException("Invalid data format:courseType is required");
        }
        if (!CourseType.contains(courseType)) {
            throw new NtClientException("courseType must be '1v1' or '1vn', courseType:" + courseType);
        }

        if (role.intValue() == UserRole.student.getCode().intValue()) {
            Integer membId = head.getMembId();
            if (membId == null) {
                throw new NtClientException("Invalid data format:membId is required");
            }
        }
        if (role.intValue() == UserRole.teacher.getCode().intValue()) {
            Integer teaId = head.getTeaId();
            if (teaId == null) {
                throw new NtClientException("Invalid data format:teaId is required");
            }
        }

        Integer roomId = head.getRoomId();
        if (roomId == null) {
            throw new NtClientException("Invalid data format:roomId is required");
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error(cause.getMessage(), cause);
        // 从教室里面移除一个人，如果这时教室已空就移除教室。
        String channelKey = ctx.channel().remoteAddress().toString();
        logger.debug("Current room size is {}", this.sessionRepository.size());
        ScheduledFuture<?> heartBeat = taskMap.get(channelKey);
        if (heartBeat != null) {
            logger.info(" \n       / cancel schedule task ------》》》》》》》》" + heartBeat.toString());
            boolean cancelSuccess = heartBeat.cancel(true);
            logger.info("channelInactive: 关闭心跳 for client:" + channelKey + " cancelHeartBeatSuccess:" + cancelSuccess + "heartBeatInstance" + heartBeat.toString());
            heartBeat = null;
        }
//        this.sessionRepository.remove(channelKey);
        logger.info("exceptionCaught: 移除客户端[{}]后Current room size is {}", channelKey, this.sessionRepository.size());
        // 释放资源
        ctx.close();
        logger.info("exceptionCaught：释放资源 for client:{}", channelKey);
        ctx.fireExceptionCaught(cause);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {

        String channelKey = ctx.channel().remoteAddress().toString();
        // 关闭心跳
        ScheduledFuture<?> heartBeat = taskMap.get(channelKey);
        if (heartBeat != null) {
            logger.info(" \n       / cancel schedule task ------》》》》》》》》" + heartBeat.toString());
            boolean cancelSuccess = heartBeat.cancel(true);
            logger.info("channelInactive: 关闭心跳 for client:" + channelKey + " cancelHeartBeatSuccess:" + cancelSuccess + ", heartBeatInstance" + heartBeat.toString());
            heartBeat = null;
        }
        this.sessionRepository.remove(channelKey);
        logger.info("channelInactive: 移除客户端[{}]后Current room size is {}", channelKey, this.sessionRepository.size());
        // 释放资源
        ctx.close();
        logger.info("channelInactive：释放资源 for client:{}", channelKey);
    }

    //Ping消息任务类
    private class HeartBeatTask implements Runnable {
        private final ChannelHandlerContext ctx;
        private volatile Session session;

        public HeartBeatTask(final ChannelHandlerContext ctx) {
            this.ctx = ctx;
            this.session = sessionRepository.getSessionByKey(ctx.channel().remoteAddress().toString());
        }

        @Override
        public void run() {

            TcpResponse heatBeat = new TcpResponse(Action.HEART_BEAT.getName());
            //发送心跳并添加一个侦听器，如果发送操作失败将关闭连接
            String heatBeatMsg = JsonUtil.toJson(heatBeat);
            heatBeat.setMessage("Ping");
            logger.info("##### endpoint[{}]status -->> isActive:{},isOpen:{},isRegistered:{},isWritable:{},", ctx.channel().remoteAddress().toString(), ctx.channel().isActive(), ctx.channel().isOpen(), ctx.channel().isRegistered(), ctx.channel().isWritable());
            ctx.channel().writeAndFlush(heatBeatMsg + Constants.delimiter).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

            String clientEndPoint = ctx.channel().remoteAddress().toString();
            if (session == null) {
                session = sessionRepository.getSessionByKey(clientEndPoint);
            }
            if (session != null) {
                User user = session.getUser();
                heatBeat.setMessage("Ping to " + user.getUserRole() + ":" + user.getName());
                logger.info("\n    =========>>>:: Send heartbeat to client[" + clientEndPoint + "] --> " + JsonUtil.toJson(user));
            } else {
                logger.info("\n    =========>>>:: Send heartbeat to client[" + clientEndPoint + "] 【None session】");
            }
        }

    }

    public void setChannelRepository(SessionRepository sessionRepository) {
        this.sessionRepository = sessionRepository;
    }
}
